package com.xiao.dynamic_rule.utils;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.Properties;

public class KafkaUtils {
    private static String keySerializer = "org.apache.kafka.common.serialization.StringSerializer";
    private static String valueSerializer = "org.apache.kafka.common.serialization.StringSerializer";

    public static KafkaProducer<String, String> getKafakProducer(String bootstrapServers) {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", bootstrapServers);
        prop.put("key.serializer", keySerializer);
        prop.put("value.serializer", valueSerializer);
        return new KafkaProducer<String, String>(prop);
    }

    public static FlinkKafkaConsumer<String> getFlinkKafkaCounsumer(String bootstrapServers, String offsetReset, String topic) {
        Properties prop = new Properties();
        prop.put("bootstrap.servers", bootstrapServers);
        prop.put("key.serializer", keySerializer);
        prop.put("value.serializer", valueSerializer);
        prop.put("auto.offset.reset", offsetReset);
        return new FlinkKafkaConsumer<String>(topic, new SimpleStringSchema(), prop);
    }
}
